package com.youxin.logcenter.kafka;



import java.util.Properties;  
  
import org.apache.kafka.clients.producer.KafkaProducer;  
import org.apache.kafka.clients.producer.ProducerRecord;  
import org.apache.kafka.clients.producer.RecordMetadata;  

import com.youxin.busfoundation.config.KafkaConfig;
  
/**
 * Kafka 发送信息工具类
 * @author yingde.cao
 *
 */
public class KafkaProducerLog {      
	private static org.apache.log4j.Logger logger = org.apache.log4j.Logger.getLogger(KafkaProducerLog.class);
	private String errorMsg=null;
	public KafkaProducerLog()
	{
		
	}
	
    public  String sendMessage(String key,String mes,KafkaConfig config) {  
        KafkaProducer<String, String> producer = null;  
        try {  
            Properties properties = new Properties();  
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
            properties.put("bootstrap.servers", config.getServers());  
            properties.put("acks", "0");  
            properties.put("retries", 0);  
            properties.put("timeout.ms", 3000);
            properties.put("socket.timeout.ms", 3000);
            /**
             * 超时时间设置
             */
            properties.put("max.block.ms", 2000);
            producer = new KafkaProducer<String, String>(properties);    
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(config.getTopics(),key, mes);  
            
            producer.send(record, (RecordMetadata recordMetadata, Exception e) -> {
				if (e != null) {
					this.errorMsg=e.toString();
					logger.error("生产消息异常", e);
				}
			});            
            return errorMsg;            
        } catch (Exception e) {  
            logger.error("Kafka发送信息异常", e);
			this.errorMsg=e.toString();
            return errorMsg;            
        } finally {  
        	try {
				if(producer!=null)
				{
		            producer.close();  
				}
			} catch (Exception e) {
	            logger.error("Kafka producer.close()", e);
			}
        }  
    }  
  
//    public static void main(String[] args) {  
//        while (true) {  
//  
//            KafkaUtil.sendMessage(  
//                    "null|127.0.0.1|国外|||||http://backserver/car/historyKeyAndHotKey.json|http://bd.2schome.net/|Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.19 Safari/537.36|http://backserver/car/historyKeyAndHotKey.json?keytype=0&|1470189803366185158");  
////          try {  
////              Thread.sleep(1000);  
////          } catch (InterruptedException e) {  
////  
////              // TODO Auto-generated catch block  
////              e.printStackTrace();  
////  
////          }  
//        }  
//  
//    }  
  
}  